package org.example.rocketmq.a01_quickstart;

import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.PullResult;
import org.apache.rocketmq.client.consumer.PullStatus;
import org.apache.rocketmq.client.consumer.store.ReadOffsetType;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.example.rocketmq.NamesrvAddr;

import java.util.HashSet;
import java.util.List;
import java.util.Set;

public class ConsumerPull {
    public static void main(String[] args) throws Exception {
        pullConsumer();
    }

    /**
     * 已过时的消息拉取
     * @throws Exception
     */
    public static void pullConsumer() throws Exception {
        DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("consumer_pull_group_01");
        consumer.setNamesrvAddr(NamesrvAddr.get());
        consumer.start();
        System.out.printf("Consumer started %n");

        Set<String> topics = new HashSet<>();
        topics.add("TopicTest");
        consumer.setRegisterTopics(topics);

        while (true) {
            consumer.getRegisterTopics().forEach(topic -> { // 遍历 topic
                try {
                    Set<MessageQueue> messageQueues =  consumer.fetchSubscribeMessageQueues(topic); // 拿到 messageQueues

                    messageQueues.forEach(messageQueue -> {
                        try {
                            long offset = consumer.getOffsetStore().readOffset(messageQueue, ReadOffsetType.READ_FROM_MEMORY); // 没有找到为 -1

                            if (offset < 0) { // 读取失败
                                offset = consumer.getOffsetStore().readOffset(messageQueue, ReadOffsetType.READ_FROM_STORE);
                            }
                            if (offset < 0) { // 读取失败
                                offset = consumer.maxOffset(messageQueue);
                            }
                            if (offset < 0) { // 读取失败
                                offset = 0;
                            }

                            // 根据偏移量获取结果
                            PullResult pullResult = consumer.pull(messageQueue, "*", offset, 32);

                            PullStatus pullStatus = pullResult.getPullStatus();
                            if (pullStatus == PullStatus.NO_NEW_MSG) {
                                return;
                            }

                            List<MessageExt> msgFoundList = pullResult.getMsgFoundList();
                            if (msgFoundList == null || msgFoundList.isEmpty()) {
                                return;
                            }
                            System.out.println("消息拉取成功： " + pullResult);

                            switch (pullResult.getPullStatus()) {
                                case FOUND:
                                    for (MessageExt messageExt : pullResult.getMsgFoundList()) {
                                        System.out.println("消息消费成功：" + messageExt);
                                        System.out.println("消息消费成功：" + new String(messageExt.getBody()));
                                    }

                                    consumer.updateConsumeOffset(messageQueue, pullResult.getNextBeginOffset()); // 更新偏移量
                                    break;
                                default:
                                    System.out.println();
                            }

                        } catch (Exception e1) {
                            e1.printStackTrace();
                        }
                    });
                } catch (MQClientException ex) {
                    ex.printStackTrace();
                }
            });
        }
    }
}
